package one.bugu.zookeeper.rpc.framework.client.socket;

import one.bugu.zookeeper.rpc.framework.client.zookeeper.IRegisterCenterInvoker;
import one.bugu.zookeeper.rpc.framework.entity.Message;
import com.google.gson.Gson;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;

/**
 * Created with IntelliJ IDEA.
 * User: LangK
 * Created Date 2019/2/14
 * Time: 10:10
 * Description:
 */
public class ClientRequestPool {

    private Logger logger = LoggerFactory.getLogger(ClientRequestPool.class);

    private static final long TIME_OUT = 3000;

    public static Map<String, ClientSocket> clientSocketMap = new ConcurrentHashMap<>();
    private Map<String, SyncFuture> promiseMap = new ConcurrentHashMap<>();

    private Gson gson = new Gson();

    private IRegisterCenterInvoker registerCenterInvoker;

    /**
     * Socket事件回调
     */
    private CallBackImpl callBack = new CallBackImpl();

    public ClientRequestPool(IRegisterCenterInvoker registerCenterInvoker) {
        this.registerCenterInvoker = registerCenterInvoker;
    }

    /**
     * 发送消息
     *
     * @param serverName 服务名
     * @param ip         服务端IP
     * @param method     方法
     * @param param      参数
     * @return
     */
    public String send(String serverName, String ip, String method, Object[] param) {
        ClientSocket clientSocket = null;
        synchronized (this) {
            try {
                String[] host = ip.split(":");
                clientSocket = clientSocketMap.get(serverName);
                if (clientSocket == null) {
                    clientSocket = getClientSocket(serverName, host);
                }
                if (clientSocket == null) {
                    return null;
                }
            } catch (Exception e) {
                logger.error("连接RPC服务端异常", e);
            }
        }
        return getResult(serverName, method, param, clientSocket);
    }

    /**
     * 发送消息
     *
     * @param serverName 服务名
     * @param method     方法
     * @param param      参数
     * @return
     */
    public String send(String serverName, String method, Object[] param) {
        List<String> providerList = registerCenterInvoker.getProviderMap().get(serverName);
        if (providerList == null || providerList.isEmpty()) {
            return null;
        }
        ClientSocket clientSocket = null;
        synchronized (this) {
            for (String provider : providerList) {
                try {
                    String[] host = provider.split(":");
                    clientSocket = clientSocketMap.get(serverName);
                    if (clientSocket == null) {
                        clientSocket = getClientSocket(serverName, host);
                        if (clientSocket != null) {
                            break;
                        }
                    } else {
                        break;
                    }
                } catch (Exception e) {
                    logger.error("连接RPC服务端异常", e);
                }
            }
        }
        if (clientSocket == null) {
            return null;
        }
        return getResult(serverName, method, param, clientSocket);
    }

    private CountDownLatch countDownLatch;

    private ClientSocket getClientSocket(String serverName, String[] host) throws InterruptedException {
        ClientSocket clientSocket = new ClientSocket(serverName, host[0], Integer.parseInt(host[1]), callBack);
        clientSocket.start();
        countDownLatch = new CountDownLatch(1);
        countDownLatch.await();
        if (clientSocketMap.containsKey(serverName)) {
            return clientSocket;
        } else {
            return null;
        }
    }

    /**
     * 获取结果
     *
     * @param serverName
     * @param method
     * @param param
     * @param clientSocket
     * @return
     */
    private String getResult(String serverName, String method, Object[] param, ClientSocket clientSocket) {
        if (clientSocket == null) {
            return null;
        }
        Message message = new Message(buildMsg(serverName, method, param));
        SyncFuture<String> syncFuture = new SyncFuture();
        this.promiseMap.put(message.getId(), syncFuture);
        boolean result = clientSocket.send(message);
        if (result) {
            try {
                syncFuture.await();
                this.promiseMap.remove(message.getId());
                return syncFuture.get();
            } catch (Exception e) {
                logger.error("发送消息异常", e);
            }
        }else {
            this.promiseMap.remove(message.getId());
        }
        return null;
    }

    /**
     * 构建消息体
     *
     * @param serverName
     * @param method
     * @param param
     * @return
     */
    private String buildMsg(String serverName, String method, Object[] param) {
        String result;
        StringBuffer stringBuffer = new StringBuffer();
        stringBuffer.append(serverName).append(">").append(method);
        if (param != null && param.length > 0) {
            stringBuffer.append(">");
            for (Object o : param) {
                stringBuffer.append(gson.toJson(o)).append("&");
            }
            result = stringBuffer.substring(0, stringBuffer.length() - 1);
        } else {
            result = stringBuffer.toString();
        }
        return result;
    }

    class CallBackImpl implements ClientSocket.CallBack {

        @Override
        public void connectionSuccess(String serverName, ClientSocket clientSocket) {
            ClientRequestPool.clientSocketMap.put(serverName, clientSocket);
            if (countDownLatch != null) {
                countDownLatch.countDown();
            }
        }

        @Override
        public void connectionFail(String serverName) {
            ClientRequestPool.clientSocketMap.remove(serverName);
            if (countDownLatch != null) {
                countDownLatch.countDown();
            }
        }

        @Override
        public void receiveMessage(Message message) {
            if (message == null) {
                return;
            }
            if (ClientRequestPool.this.promiseMap.containsKey(message.getId())) {
                ClientRequestPool.this.promiseMap.get(message.getId()).setResponse(message.getResult());
            } else {
                logger.info("丢失Feature,消息处理异常：{}", message);
            }

        }
    }


}
